package fox.framework.rpc.endpoint.netty.server;

import fox.framework.rpc.exception.*;
import fox.framework.rpc.message.RpcHeader;
import fox.framework.rpc.server.dispatcher.RpcRequestDispatcher;
import fox.framework.rpc.server.message.request.RpcServerRequestAPI;
import fox.framework.rpc.server.message.response.RpcServerResponseAPI;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import io.netty.util.concurrent.EventExecutorGroup;

/**
 * @author cuichao
 * @Description: RpcServerDispatcherHandler
 * @Date: create in 2021/1/20 20:53
 */
@ChannelHandler.Sharable
public class RpcServerDispatcherHandler extends SimpleChannelInboundHandler<RpcDecodeServerRequest> {

    private EventExecutorGroup executors = new DefaultEventExecutorGroup(16);
    private RpcRequestDispatcher dispatcher;

    public RpcServerDispatcherHandler(RpcRequestDispatcher dispatcher) {
        this.dispatcher = dispatcher;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcDecodeServerRequest request) {
        executors.submit(() -> dispatch(ctx, request));
    }

    private void dispatch(ChannelHandlerContext ctx, RpcDecodeServerRequest request) {
        //wrapper to api
        RpcServerRequestAPI requestAPI = new RpcServerRequestAPI(request);
        RpcEncodeServerResponse response = new RpcEncodeServerResponse(request.getProtocol());
        RpcServerResponseAPI responseApi = new RpcServerResponseAPI(response);

        dispatcher.dispatch(requestAPI, responseApi);
        //decode
        response.decode();
        //write

        ctx.writeAndFlush(response).addListener(future -> {
           if (!future.isSuccess()) {
                Throwable cause = future.cause();
                exceptionHandler(ctx, cause);
            }
        });

    }

    /**
     * 处理读取异常
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        exceptionHandler(ctx, cause);
    }


    private void exceptionHandler(ChannelHandlerContext ctx, Throwable cause){
        Throwable exception = findRpcRuntimeException(cause);
        if(exception == null){
            //其他异常
            cause.printStackTrace();
            return;
        }
        /**
         * 操作记录形异常
         */
        if (exception instanceof RpcRecordException) {
            exception.printStackTrace();
        }
        /**
         * 可发送异常
         */
        if(exception instanceof RpcDecodeWithProtocolException){
            RpcEncodeServerResponse response = new RpcEncodeServerResponse(((RpcDecodeWithProtocolException) exception).getProtocol());
            if(exception instanceof RpcDecodeWithRequestIdException){
                response.setHeader(RpcHeader.REQUEST_ID, ((RpcDecodeWithRequestIdException) exception).getRequestId());
            }
            dispatcher.processExceptionHandler(response,exception);
            response.decode();
            ctx.writeAndFlush(response).addListener(future -> {
                if(future.cause() != null){
                    cause.printStackTrace();
                }

            });
            return;
        }

    }


    /**
     * 查找rpcRuntime异常类
     *
     * @param cause
     * @return
     */
    private Throwable findRpcRuntimeException(Throwable cause) {
        if (cause == null) {
            return null;
        } else if (cause instanceof RpcRuntimeException) {
            return cause;
        } else {
           return findRpcRuntimeException(cause.getCause());
        }
    }



}
